	// Copyright (c) 2007 Intel Corp. (c) 2009 Daniele Buono
//

// Black-Scholes
// Analytical method for calculating European Options
//
//
// Reference Source: Options, Futures, and Other Derivatives, 3rd Edition, Prentice
// Hall, John C. Hull,


#include "cycle.h"
#include "channel.h"

#include "blkschlsfunc.h"

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>	// ftok
#include <unistd.h>		// fork
#include <string.h>		// memcpy
#include <sys/wait.h>	// waitpid

#include <pthread.h>

#ifdef AFFINITY
#include <sched.h>
#endif

#define MAX_THREADS 128

#define PAD 256
#define LINESIZE 64

fptype* prices;
int numOptions;
int nThreads;
int nRuns;

typedef struct ThreadData_ {
        int id;
        key_t inkey;
        key_t outkey;
        key_t *keys;
        key_t resultkey;
        int	*elements;
} ThreadData;

void emitter(OptionData* data, int keys[]){
	int * otype;
	int i,j;
	fptype * sptprice;
	fptype * strike;
	fptype * rate;
	fptype * volatility;
	fptype * otime;

	void* tempbuffer[3];

	Channel *out = malloc(sizeof(Channel)*nThreads);

	// Prepare Channels
	for (i = 0; i < nThreads; i++) {
		out[i] = LCattach(keys[i]);
	}

	for (j = 0; j < nRuns; j++) {

		// Data is allocated exactly as in the original PARSEC version
		tempbuffer[0] = (int *) malloc(numOptions * sizeof(fptype) + PAD);
		otype = (int *) (((unsigned long long) tempbuffer[0] + PAD) & ~(LINESIZE - 1));

		tempbuffer[1] = (int *) malloc(5 * numOptions * sizeof(fptype) + PAD);
		sptprice = (fptype *) (((unsigned long long) tempbuffer[1] + PAD)
				& ~(LINESIZE - 1));
		strike = sptprice + numOptions;
		rate = strike + numOptions;
		volatility = rate + numOptions;
		otime = volatility + numOptions;

		tempbuffer[2] = (int*) malloc(numOptions * sizeof(fptype));


		for (i = 0; i < numOptions; i++) {
			otype[i] = (data[i].OptionType == 'P') ? 1 : 0;
			sptprice[i] = data[i].s;
			strike[i] = data[i].strike;
			rate[i] = data[i].r;
			volatility[i] = data[i].v;
			otime[i] = data[i].t;
		}

		// Send the address of allocated buffers to workers
		for(i=0; i<nThreads; i++){
		    printf("-Emitter:invio i dati a %d\n",i);
			LCsend(out[i],(char*)tempbuffer);
		}
	}

	// Send the EOF
	tempbuffer[0] = NULL;
	for(i=0; i<nThreads; i++){
		LCsend(out[i],(char*)tempbuffer);
	}

	printf("Emitter: done\n");

	// Detach Channels
	for (i = 0; i < nThreads; i++) {
		LCdetatch(out[i]);
	}
}

int collector(key_t keys[], key_t resultkey){
	int i;
	fptype *data;
	fptype ***buffers = malloc(sizeof(void*)*nThreads);
	Channel *in = malloc(sizeof(Channel)*nThreads);
	Channel result;

	// Prepare Channels
	result = LCattach(resultkey);
	for (i = 0; i < nThreads; i++) {
		in[i] = LCattach(keys[i+nThreads]);
	}

	// This is not to receive N copies of the same pointer, but to know when a task is completed
	for(i=0; i<nThreads; i++){
		buffers[i] = (fptype**) LCreceive(in[i]);
	}

	data = NULL;

	// Start receiving from workers
	while(buffers[0][0] != NULL ){
		if(data!=NULL)
			free(data);
		free(buffers[0][0]);
		free(buffers[0][1]);
		data = buffers[0][2];

		for(i=0; i<nThreads; i++){
		    printf("-Collector: ricevo dati da %d...\n",i);
			buffers[i] = (void*) LCreceive(in[i]);
			printf("-Collector: ricevuti dati da %d\n",i);
		}
	}

	printf("Collector: done\n");

	// Send back the last received message to the parent thread
	LCsend(result,(char*)&data);

	// Detach Channels
	for (i = 0; i < nThreads; i++) {

		LCdestroy(in[i]);
	}
	LCdetatch(result);

	return 0;
}

int worker(int id, key_t inkey, key_t outkey, int* elements) {

	int i, k;
	fptype price[NCO];

	int start,end;
	start=0;
	for(i=1;i<=id;i++){
		start+=elements[i-1];
	}
	end = start + elements[id];

	printf("My tid is %u. Start: %d. End: %d\n", pthread_self(), start, end);

#ifdef AFFINITY
	cpu_set_t mask;
	/* CPU_ZERO initializes all the bits in the mask to zero. */
	CPU_ZERO(&mask);
	/* CPU_SET sets only the bit corresponding to cpu. */
	CPU_SET(id, &mask);
	/* sched_setaffinity returns 0 in success */
	if (sched_setaffinity(0, sizeof(mask), &mask) == -1)
	{
		printf("WARNING: Could not set CPU Affinity, continuing...\n");
	}
#endif

	int    * otype;
	fptype * sptprice;
	fptype * strike;
	fptype * rate;
	fptype * volatility;
	fptype * otime;
	fptype *prices;
	void** data = NULL;


	// Attach to the channel
	Channel in = LCattach(inkey);
	Channel out = LCattach(outkey);

	// Receive the first message
	printf("- worker %d, ricevo i primi dati dall emitter\n",id);
	data = (void**) LCreceive(in);
	while(data[0]!= NULL){

		sptprice = (fptype *) (((unsigned long long)data[1] + PAD) & ~(LINESIZE - 1));
		strike = sptprice + numOptions;
		rate = strike + numOptions;
		volatility = rate + numOptions;
		otime = volatility + numOptions;
		otype = (int *) (((unsigned long long)data[0] + PAD) & ~(LINESIZE - 1));

		prices = data[2];

		for (i=start; i<end; i += NCO) {
			// Calling main function to calculate option value based on Black & Scholes's
			// equation.
			BlkSchlsEqEuroNoDiv(price, NCO, &(sptprice[i]), &(strike[i]),
					&(rate[i]), &(volatility[i]), &(otime[i]), &(otype[i]), 0);
			for (k=0; k<NCO; k++) {
				prices[i+k] = price[k];
			}
		}

		// Inform the collector
		printf("- worker %d, invio i dati al collector\n",id);
		LCsend(out, (char*) data);

		// Receive the next one
		printf("- worker %d, ricevo i dati dall emitter\n",id);
		data = (void**) LCreceive(in);
	}

	// Forward to the collector the EOF signal
	printf("-worker %d, EOF\n",id);
	LCsend(out, (char*) data);

printf("Worker %d. Closing...\n", id);
	LCdestroy(in);
	LCdetatch(out);



	return 0;
}


int startThread(void *tid_ptr){
	ThreadData *data = (ThreadData*) tid_ptr;
	if(data->id<nThreads){	// This is a worker
		return worker(data->id,data->inkey,data->outkey,data->elements);
	}else{	// This is the collector
		return collector(data->keys,data->resultkey);
	}
}

int main(int argc, char **argv) {
	int i;

	OptionData* data;
	fptype **results;

	printf("BlackScholes Benchmark - Messages Version\n");
	fflush(NULL);

	if (argc != 5) {
		printf("Usage:\n\t%s <nThreads> <nruns> <inputFile> <outputFile>\n",
				argv[0]);
		exit(1);
	}

	nThreads = atoi(argv[1]);
	nRuns = atoi(argv[2]);
	char *inputFile = argv[3];
	char *outputFile = argv[4];

	printf("This cpu have a clockrate of %f Mhz\n", (double) CLOCKRATE / 1000000.0);

	ticks clocks;

	data = loadData(inputFile,&numOptions,&nThreads);

	printf("Num of Options: %d\n", numOptions);
	printf("Num of Runs: %d\n", nRuns);

	printf("Size of data: %ld\n", numOptions * (sizeof(OptionData)
			+ sizeof(int)));

	// Initialization...

	// Define partition sizes
	int elemperworker = (numOptions / nThreads);
	while(elemperworker%NCO != 0 ){
		elemperworker++;
	}

	Channel tmpchan;
	key_t *keys = malloc(sizeof(key_t) * nThreads * 2 );
	int	*elements = malloc(sizeof(int) * nThreads);
	ThreadData tdata[MAX_THREADS+1];


	/*
	 * I create the channels here so they exists when workers and emitter starts
	 * However they are immediately detached so they are not retained after forks
	 */
	for (i = 0; i < nThreads; i++) {
		elements[i] = elemperworker;
		if(i==nThreads-1)
			elements[i] = numOptions - ((nThreads-1) * elemperworker);

		keys[i] = ftok(argv[0],i);
		keys[i+nThreads] = ftok(argv[0],i+nThreads);
		if(keys[i]==-1){
			printf("Error creating a new key with parameters %s %d\n",argv[0],i);
			exit(-1);
		}
		tmpchan =  LCcreate_as( keys[i], sizeof(void*) * 3 , 1, 1 );
		LCdetatch(tmpchan);
		tmpchan =  LCcreate_as( keys[i+nThreads], sizeof(void*) * 3, 1, 1 );
		LCdetatch(tmpchan);

		// Fill tdata
		tdata[i].id=i;
		tdata[i].inkey=keys[i];
		tdata[i].outkey=keys[i+nThreads];
		tdata[i].elements=elements;
	}
	key_t resultkey = ftok(argv[0],2*nThreads);
	tmpchan = LCcreate_ss( resultkey, sizeof(void*), 1 );
	LCdetatch(tmpchan);

	tdata[i].id=i;
	tdata[i].keys=keys;
	tdata[i].resultkey=resultkey;

	// Start workers
	pthread_mutexattr_t normalMutexAttr;
	pthread_mutexattr_init(&normalMutexAttr);

	pthread_t tids[MAX_THREADS+1];
	for (i = 0; i <= nThreads; i++) {
		pthread_create(&tids[i], NULL, (void *(*)(void *)) startThread,
			(void *) &tdata[i]);
	}

	Channel resultc = LCattach(resultkey);

	// Let's start the computation!
	clocks = getticks();

	// Act as an emitter
	emitter(data,keys);

	results = (fptype**) LCreceive(resultc);
	// The computation is done!

	clocks = getticks() - clocks;

	free(elements);
	free(keys);

	// Wait for the threads to quit
	for (i = 0; i < nThreads; i++) {
		void *ret;
		pthread_join(tids[i], &ret);
	}

	printf("Total computation time: %lf\n", ((double)clocks)/CLOCKRATE);

	printf("Service time: %f\n", ((double)clocks)/ CLOCKRATE / nRuns);

	// Save on times.txt
	FILE *times = fopen("times.txt","a");
	if (times == NULL) {
		printf("ERROR: Unable to open file %s.\n", "times.txt");
		exit(1);
	}
	fprintf(times,"%d\t\t%f\n",nThreads,((double)clocks)/CLOCKRATE );
	fclose(times);

	printf("Starting saveData\n");
	fflush(stdout);
	saveData(*results,outputFile,numOptions);

/*	free(results[2]);

	free(data);*/
	LCdestroy(resultc);

	return 0;
}
